package com.crazyice.lee.jobs;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.table.sinks.CsvTableSink;
import org.apache.flink.table.sinks.TableSink;
import org.apache.flink.table.sources.CsvTableSource;
import org.apache.flink.table.sources.TableSource;
import org.apache.flink.types.Row;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

@Slf4j
public class BatchTableSQLJob {
    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = BatchTableSQLJob.class.getResourceAsStream("/file.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    public static void main(String[] args) throws Exception {
        init();

        //设置环境
        final ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        final BatchTableEnvironment tableEnv = BatchTableEnvironment.getTableEnvironment(environment);

        //源数据
        TypeInformation[] fieldTypes = {Types.INT, Types.STRING, Types.LONG};
        String[] fieldName={"id","name","age"};

        TableSource csvSource = new CsvTableSource(properties.getProperty("source"),fieldName,fieldTypes);
        tableEnv.registerTableSource("source", csvSource);

        TableSink csvSink = new CsvTableSink(properties.getProperty("sink"),",",1, FileSystem.WriteMode.OVERWRITE);
        tableEnv.registerTableSink("sink",fieldName,fieldTypes, csvSink);

        //处理过程
        Table dest=tableEnv.sqlQuery("select * from source where age>20").orderBy("age.asc").fetch(10);

        //结果输出-table转换为DataSet，再实现output接口输出
        dest.insertInto("sink");
        DataSet<Row> result = tableEnv.toDataSet(dest, Row.class);
        result.output(new OutputFormat<Row>() {
            @Override
            public void configure(Configuration configuration) {

            }

            @Override
            public void open(int i, int i1) throws IOException {

            }

            @Override
            public void writeRecord(Row row) throws IOException {
                log.info("计算结果:{}",row);
            }

            @Override
            public void close() throws IOException {

            }
        }).name("输出到日志");

        environment.execute("流和维表Join演示");
    }
}
